package com.o2o.cleaning.month.platform.ebusiness_plat.brand_modular

import com.alibaba.fastjson.JSON
import org.apache.commons.codec.digest.DigestUtils
import org.apache.spark.sql.{DataFrame, SparkSession}


object brand_join_res_new {
  /**
   * @param sourceData      :DataFrame  跑完分类的【数据】
   * @param BrandResultPath ：String  跑完品牌的数据结果路径
   * @param year            ：int                 年
   * @param month           ：int                月
   * @param platformName    ：string     平台名称
   * @param spark
   */

  def brandJoinResult(sourceData: DataFrame, BrandResultPath: String,
                      year: Int, month: Int, platformName: String, spark: SparkSession) {
    // ************************************************************************************************************
    val timeStamp = s"${year}年${month}月"
    val brandTableResult = s"s3a://o2o-dimension-table/brandName_table/${year}/${month}/${platformName}/"
    val brandNewAddTable = s"s3a://o2o-dimension-table/newAddBrand/${year}/${month}/${platformName}/"
    // ************************************************************************************************************

    println("--------------提取新增品牌--------------------------")
    sourceData.selectExpr(s"'${platformName}' platformName", "brandValueId", "brandName", "firstCategoryId").dropDuplicates("brandValueId")
      .registerTempTable("sourcedata")
    //    sourceData.withColumnRenamed("brand_name","brandName").withColumnRenamed("brand_id","brandValueId").selectExpr(s"'${platformName}' platformName", "brandValueId", "brandName", "firstCategoryId")
    //      .registerTempTable("sourcedata")

    //    读取品牌表
    val new_brand = spark.sqlContext.sql(
      s"""
         |select
         |'${timeStamp}'timeStamp,
         |t1.brandCcId,
         |t1.platformName,
         |t1.brandValueId,
         |t1.brandName,
         |t1.firstCategoryId
         |from
         |(select *,
         |concat(brandValueId,"|",firstCategoryId) brandCcId
         |from sourcedata
         |) t1
      """.stripMargin)
      //      将没有的字段补全
      .na.fill(Map("brandCcId" -> "-1"))
      .dropDuplicates("brandCcId")
      .where("brandName != '-1' and brandValueId != '-1' ")

    //    统计一下新增品牌条数，如果没有新品牌直接关联品牌入库
    val brandNewCount = new_brand.dropDuplicates("brandCcId").count()
    if (brandNewCount > 0) {
      val brandNewAddToOld = new utils.BrandNewAddToOld
      brandNewAddToOld.newToOld_newplat(new_brand, spark, brandTableResult,
        BrandResultPath, sourceData, platformName, year, month, brandNewAddTable, timeStamp)
    }
  }


  def main(args: Array[String]): Unit = {
    val spark = SparkSession.builder().appName("Other")
      .master("local[*]")
      .config("es.nodes", s"192.168.1.29")
      .config("es.port", "9200")
      .config("cluster.name", "O2OElastic")
      .getOrCreate()
    spark.sparkContext.hadoopConfiguration.set("fs.s3a.access.key", "GAO7EO9FWKPJ8WFCQDME")
    spark.sparkContext.hadoopConfiguration.set("fs.s3a.secret.key", "LZ0xaHBSYKHaJ9ECDbX9f7zin79UZkXfGoNapRPL")
    spark.sparkContext.hadoopConfiguration.set("fs.s3a.endpoint", "https://obs.cn-north-1.myhuaweicloud.com")
    val frame = spark.read.json("s3a://o2o-dataproces-group/tian_yaxin/douyin/douyinResult/").drop("add_to_field").toJSON.rdd.map(li => {
      val jsons = JSON.parseObject(li)
      val brandName = jsons.getOrDefault("brandName","-1").toString
      val brandValueId = DigestUtils.md5Hex(brandName.toUpperCase())
      jsons.put("brandValueId", brandValueId)
      jsons.toString
    })
    val platform = "douyin"
    val frame1 = spark.read.json(frame)
    frame1
      .printSchema()
    val data_path = ""
    brandJoinResult(frame1, data_path, 2021, 4, platform, spark)
    spark.stop()
  }
}


